package udf.matching;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

public class JoinUsingBagComplex extends EvalFunc<DataBag> {

	static private boolean initialized = false;

	private BagFactory bagFactory = BagFactory.getInstance();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static HashSet<String> seeds = new HashSet<String>();
	static private Hashtable<String, HashSet<String>> seeds = new Hashtable<String, HashSet<String>>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";
	/**
	 * Tuple has the following fields
	 * 
	 * 
	 * 0 - tuple with fields to be intersected
	 * 
	 * 1 - dictionary to do the look up
	 * 
	 * 2 - position of tag in dictionary of (1)
	 * 
	 * 3 - resorting of tuples in , 3 spicify the position in 1 4- tuple with
	 * the positions to look at
	 * 
	 * 
	 * 
	 */

	/**
	 * in this implementaion is assumed:
	 * 
	 * tag to match always ins pos 0 query of second thing in pos 1
	 * 
	 */
	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(1);

		String tag_pos_ = (String) tuple.get(2);
		int tag_pos = Integer.valueOf(tag_pos_);

		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();
			
			String tag = (String) tup.get(tag_pos);
			String query = (String) tup.get(1);
			// System.out.println("Adding: " + tag + "\t" + temp[0]);

			if (seeds.containsKey(tag)) {

				HashSet<String> temporal = seeds.get(tag);
				temporal.add(query);
				
				seeds.put(tag, temporal);
				
			//	System.out.println("Adding to dict:" + tag+ "\t "  + query);

			} else {
				HashSet<String> temporal = new HashSet<String>();
				temporal.add(query);
				seeds.put(tag, temporal);
				
				//System.out.println("Adding to dict:" + tag+ "\t "  + query);
			}

		}

		initialized = true;
	}

	public DataBag exec(Tuple input) throws IOException {

		if (!initialized)
			init(input);

		Tuple tags = (Tuple) input.get(0);
		String pos_ = (String) input.get(3);

		Tuple indexes = (Tuple) input.get(4);

		int debug = 0;

		if (input.size() > 5) {
			debug = (Integer) input.get(5);

		}
		int pos = Integer.valueOf(pos_);

		DataBag output = bagFactory.newDefaultBag();

		if (debug == 1)
			System.out.println("matching:");
		for (int i = 0; i < indexes.size(); i++) {
			int index = (Integer) indexes.get(i);
			String tag = (String) tags.get(index);
			if (debug == 1)
				System.out.println("\t" + tag);
			
			
			if (seeds.containsKey(tag)) {

				// Found , add tag,query rows to the output
				String temporal = (String) tags.get(pos);

				if (debug == 1)
					System.out.println("\t" + "found position:" + i);
				
				//System.out.println("\t" + "found position:" + index + "\t" + tag);

				HashSet<String> queries = seeds.get(tag);

				Iterator<String> iterator = queries.iterator();
				

				while (iterator.hasNext()) {
					String query = iterator.next();

					Tuple t = TupleFactory.getInstance().newTuple();
					for (int j = 0; j < tags.size(    ); j++) {
						if (j == pos) {
							// t.set(i, target);
							t.append(tag);

						} else if (j == index) {
							// t.set(i, temporal);
							t.append(temporal);

						} else {
							// t.set(i, tags.get(i));
							t.append( tags.get(j));
						}

					}
					
					t.append(query);
					t.append(index);
					output.add(t);

				}

				// System.out.println("Matching " + tag);
			}

		}
		if(output.size()>0)
		return output;
		else
			return null;

	}
}